package com.fubowen.reader.book.impl

import androidx.annotation.IntDef
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.launch
import kotlinx.coroutines.sync.Semaphore
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.PriorityBlockingQueue
import java.util.concurrent.atomic.AtomicInteger

class Downloader {

    //region 协程数目控制

    private var coroutineCount = MutableStateFlow(1)
    var coroutineCountFlow = coroutineCount.asStateFlow()
    private var semaphore = Semaphore(coroutineCount.value)


    fun setCoroutineCount(count: Int) {
        coroutineCount.value = count
        semaphore = Semaphore(count)
    }

    //endregion

    //region 任务运行控制
    private val waitingQueue = PriorityBlockingQueue<Task>()
    private val runningQueue = ConcurrentHashMap<Task, Job>()

    private var groupId = AtomicInteger(0)
    private var groupConfig = ConcurrentHashMap<Int, TaskGroup>()

    val onStateChanged: (Task) -> Unit = {}

    private fun requestRunning(task: Task, onCompleted: () -> Unit): Boolean {
        if (semaphore.tryAcquire()) {
            task.status = Task.STATUS_RUNNING
            onStateChanged(task)
            val job = CoroutineScope(Dispatchers.IO).launch {
                val taskGroup = groupConfig.getOrPut(task.groupId) { TaskGroup() }
                try {
                    task.invoke()
                    task.status = Task.STATUS_SUCCESS
                    taskGroup.success.incrementAndGet()
                } catch (_: CancellationException) {
                } catch (e: Exception) {
                    task.status = Task.STATUS_ERROR
                    task.message = e.message
                    taskGroup.error.incrementAndGet()
                } finally {
                    taskGroup.completed.incrementAndGet()
                    taskGroup.onCompleted(taskGroup)
                    onStateChanged(task)
                }
            }
            job.invokeOnCompletion {
                runningQueue.remove(task)
                semaphore.release()
                onCompleted()
            }
            runningQueue[task] = job
            return true
        }
        return false
    }

    private fun startNext() {
        repeat(semaphore.availablePermits) {
            waitingQueue.poll()?.apply {
                requestRunning(this) { startNext() }
            }
        }
    }

    fun start(item: Task) {
        if (!requestRunning(item) { startNext() }) {
            waitingQueue.offer(item)
        }
    }

    fun start() = startNext()

    fun startGroup(taskList: List<Task>, onTaskCompleted: (TaskGroup) -> Unit = { }): Int {
        val groupId = this.groupId.incrementAndGet()
        val taskGroup = groupConfig.getOrPut(groupId) { TaskGroup() }
        taskGroup.onCompleted = onTaskCompleted
        taskGroup.total = taskList.size
        taskList.forEach {
            it.groupId = groupId
            start(it)
        }
        return groupId
    }

    fun stop(item: Task) = runningQueue.remove(item)?.cancel()
    //endregion
}

//region Task
class Task(
    var id: String = "",
    var groupId: Int = -1,
    var priority: Int = 0,
    @Status
    var status: Int = STATUS_WAITING,
    var message: String? = "",
    var invoke: () -> Unit
) : Comparable<Task> {
    companion object {

        const val STATUS_WAITING = 0
        const val STATUS_RUNNING = 1
        const val STATUS_PAUSED = 2
        const val STATUS_ERROR = 3
        const val STATUS_SUCCESS = 4
    }

    @IntDef(STATUS_WAITING, STATUS_RUNNING, STATUS_PAUSED, STATUS_ERROR, STATUS_SUCCESS)
    @Retention(AnnotationRetention.SOURCE)
    annotation class Status

    override fun compareTo(other: Task): Int {
        return priority.compareTo(other.priority)
    }
}
//endregion

//region TaskGroup
class TaskGroup(
    var id: Int = 0,
    var total: Int = 0,
    var success: AtomicInteger = AtomicInteger(0),
    var error: AtomicInteger = AtomicInteger(0),
    var completed: AtomicInteger = AtomicInteger(0),
    var onCompleted: (TaskGroup) -> Unit = {}
)
//endregion

fun <T> T.toTask(block: T.() -> Unit) = Task { block() }

fun <T> List<T>.toTaskList(block: T.() -> Unit) = this.map { it.toTask { block() } }